package com.edu.learning.concurrent.sample.mode4;

import java.util.Comparator;
import java.util.Random;
import java.util.concurrent.*;

import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * 优先级队列PriorityQueue有以下几个特点：
 * 1. 队列中的元素总是按照“自然顺序”进行排序，或者根据构造函数中给定的Comparator进行排序；
 * 2. 队列中不允许存在null，也不允许存在不能排序的元素；
 * 3. 对于排序值相同的元素，其序列是不保证的，当然你可以自己扩展这个功能；
 * 4. 队列容量是没有上限的，但是如果插入的元素超过负载，有可能会引起OutOfMemory异常；
 * 5. 使用迭代子iterator()对队列进行轮询，其顺序不能保证；
 * 那么PriorityBlockingQueue除了以上特点外，还有另外的特点：
 * 6. 它还具有BlockingQueue的put和take方法，但是由于队列容量没有上线，所以put方法是不会被阻塞的，但是take方法是会被阻塞的；
 * 7. 可以给定初始容量，这个容量会按照一定的算法自动扩充。
 *
 * 例子中定义了一个按照字符串倒序排列的队列
 * 5个生产者不断产生随机字符串放入队列，5个消费者不断从队列中取出随机字符串，可以看到若队列为空，取出的线程会等待；
 * 也可以看到同一个线程取出的字符串基本上是倒序的（因为不同线程同时存取元素，因此取出的字符串打印到屏幕上往往不是倒序的了）
 * 在需要使用优先级队列，且还有阻塞需求时，就可以使用PriorityBlockingQueue
 *
 * @author ll
 * @Date 2017/10/26 14:08
 */
public class PriorityBlockingQueueDemo {
    public static void main(String[] args) {
        PriorityBlockingQueue<String> queue = new PriorityBlockingQueue<>(3, new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                if (o1.compareTo(o2) < 0) {
                    return 1;
                } else if (o1.compareTo(o2) > 0) {
                    return -1;
                } else {
                    return 0;
                }
            }
        });

        ExecutorService executorService = new ThreadPoolExecutor(
                32,
                32,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Producer("Product " + i, queue));
        }
        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Consumer("Consumer " + i, queue));
        }
        executorService.shutdown();
    }

    static class Producer implements Runnable {

        private final PriorityBlockingQueue<String> queue;
        private final String name;
        private static Random rand = new Random(System.currentTimeMillis());

        public Producer(String name, PriorityBlockingQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 10; i ++) {
                    String str = "Product" + rand.nextInt(1000);
                    queue.put(str);
                    sb.append("-->").append(name).append(" product ").append(str).append(", queue.size = ").append(queue.size());
                    System.out.println(sb);//name + " product " + sb + "\t , queue.size = " + blockingQueue.size()
                    SECONDS.sleep(rand.nextInt(5));
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {

        private final PriorityBlockingQueue<String> queue;
        private final String name;
        private static Random rand = new Random(System.currentTimeMillis());

        public Consumer(String name, PriorityBlockingQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 10; i ++) {
                    String str = queue.take();
                    sb.append("<--").append(name).append(" consume ").append(str).append(", queue size = ").append(queue.size());
                    System.out.println(sb);
                    SECONDS.sleep(rand.nextInt(5));
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
